package com.titan.single.kafka;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import java.util.concurrent.CompletableFuture;

/**
 * 生产者
 */
@Service
@Slf4j
public class KafkaProducer {

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    /**
     * 发送简单文本消息-异步
     */
    public void sendMessage(String topic, String message) {
        try {
            CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    log.info("生产者--发送消息成功: [{}] with offset=[{}]", message, result.getRecordMetadata().offset());
                } else {
                    log.error("生产者--发送消息失败: [{}] due to: {}", message, ex.getMessage());
                }
            });
        } catch (Exception e) {
            log.error("生产者--发送消息异常: {}", e.getMessage(), e);
        }
    }

    /**
     * 发送简单文本消息-同步
     */
    public void sendMessageSync(String topic, String message) throws Exception {
        try {
            CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, message);
            SendResult<String, String> result = future.get(); // 阻塞等待结果
            log.info("发送消息成功: [{}] with offset=[{}]", message, result.getRecordMetadata().offset());
        } catch (Exception e) {
            log.error("发送消息失败: {}", e.getMessage(), e);
            throw e;
        }
    }



    /**
     * 发送带key的消息-异步
     */
    public void sendMessageWithKey(String topic, String key, String message) {
        try {
            CompletableFuture<SendResult<String, String>> future = kafkaTemplate.send(topic, key, message);
            future.whenComplete((result, ex) -> {
                if (ex == null) {
                    log.info("发送消息成功: key=[{}], message=[{}] with offset=[{}]", 
                            key, message, result.getRecordMetadata().offset());
                } else {
                    log.error("发送消息失败: key=[{}], message=[{}] due to: {}", 
                            key, message, ex.getMessage());
                }
            });
        } catch (Exception e) {
            log.error("发送消息异常: {}", e.getMessage(), e);
        }
    }


}